package com.vic.publish_confirm;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmCallback;
import com.vic.utils.RabbitmqUtils;

import java.nio.charset.StandardCharsets;
import java.util.UUID;
import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;

/**
 * 发布确认
 */
public class Producer {


    public static void main(String[] args) throws Exception {
        // 1 单个发布确认
        // 单个发布确认，发送1000条消息，共耗时: 24417ms
//        publishMsgIndividually();

        // 2 批量发布确认
        // 批量发布确认，发送1000条消息，共耗时: 795ms
//        publishMsgBatch();

        // 3 异步发布确认，性能最高
        // 异步发布确认，发送1000条消息，共耗时: 122ms
        publishMsgAsync();

    }

    // 3 异步发布确认，性能最高
    private static void publishMsgAsync() throws Exception {
        Channel channel = RabbitmqUtils.getChannel();
        // 开启发布确认
        channel.confirmSelect();

        /**
         * 线程安全有序的一个哈希表，适用于高并发的情况
         * 1.轻松的将序号与消息进行关联
         * 2.轻松批量删除条目，只要给到序列号
         * 3.支持并发访问
         */
        ConcurrentSkipListMap<Long, String> outstandingConfirms = new ConcurrentSkipListMap<>();

        // 队列
        String queueName = UUID.randomUUID().toString();
        channel.queueDeclare(queueName, false, false, false, null);

        // 发送消息总条数
        int msgCount = 1000;

        /**
         * broker消息确认成功的回调函数
         * 1. 消息标记
         * 2. 是否批量确认
         */
        ConfirmCallback ackCallback = (sequenceNumber, multiple) -> {
            System.out.println("确认收到消息，sequenceNumber: " + sequenceNumber + ", multiple: " + multiple);
            // 批量确认
            if (multiple) {
                // 返回的是小于等于当前序列号的未确认消息 是一个 map
                ConcurrentNavigableMap<Long, String> confirmed = outstandingConfirms.headMap(sequenceNumber, true);
                // 清除该部分未确认消息
                confirmed.clear();
            } else {
                // 单个确认
                // 只清除当前序列号的消息
                outstandingConfirms.remove(sequenceNumber);
            }
        };

        /**
         * broker消息确认失败的回调函数
         * 可以进行业务处理
         */
        ConfirmCallback nackCallback = (sequenceNumber, multiple) -> {
            String message = outstandingConfirms.get(sequenceNumber);
            System.out.println("发布的消息" + message + "未被确认，序列号: " + sequenceNumber);
        };

        /**
         * 添加监听器
         * 1. 监听哪些消息成功了
         * 2. 监听哪些消息失败了
         */
        channel.addConfirmListener(ackCallback, nackCallback);

        long start = System.currentTimeMillis();

        // 发送1000条消息
        for (int i = 1; i <= msgCount; i++) {

            String message = "lure" + i;
            /**
             * 记录所有要发送的消息
             * channel.getNextPublishSeqNo()获取下一个消息的序列号
             * 通过序列号与消息体进行一个关联
             * 全部都是未确认的消息体
             */
            outstandingConfirms.put(channel.getNextPublishSeqNo(), message);

            // 发布消息
            channel.basicPublish("",queueName, null, message.getBytes(StandardCharsets.UTF_8));
        }

        long end = System.currentTimeMillis();
        System.out.println("异步发布确认，发送" + msgCount + "条消息，共耗时: " + (end - start) + "ms");
    }

    // 2 批量发布确认
    private static void publishMsgBatch() throws Exception {
        Channel channel = RabbitmqUtils.getChannel();
        // 开启发布确认
        channel.confirmSelect();

        // 队列
        String queueName = UUID.randomUUID().toString();
        channel.queueDeclare(queueName, false, false, false, null);

        // 发送消息总条数
        int msgCount = 1000;

        long start = System.currentTimeMillis();

        // 一次确认的数量
        int batchSize = 100;

        // 统计未确认的消息
        int outstandingMsgCount = 0;

        // 发送1000条消息
        for (int i = 1; i <= msgCount; i++) {
            // 发布消息
            channel.basicPublish("",queueName, null, ("aa" + i).getBytes(StandardCharsets.UTF_8));

            // 消息
            outstandingMsgCount++;

            // 接收broker的确认
            // 隔100条消息，批量确认一次
            if(i % batchSize == 0) {
                // 重置为0
                outstandingMsgCount = 0;
                boolean suc = channel.waitForConfirms();
                if(suc) {
                    System.out.println("批量确认-消息发送成功");
                }
            }
        }
        // 为了确保还有剩余没有确认消息，再次确认
        if(outstandingMsgCount > 0) {
            channel.waitForConfirms();
        }

        long end = System.currentTimeMillis();
        System.out.println("批量发布确认，发送" + msgCount + "条消息，共耗时: " + (end - start) + "ms");
    }

    // 1 单个发布确认
    private static void publishMsgIndividually() throws Exception {
        Channel channel = RabbitmqUtils.getChannel();
        // 开启发布确认
        channel.confirmSelect();

        // 队列
        String queueName = UUID.randomUUID().toString();
        channel.queueDeclare(queueName, false, false, false, null);

        // 发送消息总条数
        int msgCount = 1000;

        long start = System.currentTimeMillis();

        // 发送1000条消息
        for (int i = 1; i <= msgCount; i++) {
            // 发布消息
            channel.basicPublish("",queueName, null, ("aa" + i).getBytes(StandardCharsets.UTF_8));

            // 接收broker的确认
            // 如果服务端返回 false 或 超时时间内未返回，生产者这里可以加个消息重发的逻辑
            boolean suc = channel.waitForConfirms();
            if(suc) {
                System.out.println("消息发送成功");
            }
        }
        long end = System.currentTimeMillis();
        System.out.println("单个发布确认，发送" + msgCount + "条消息，共耗时: " + (end - start) + "ms");
    }


}
